package io.kiki.sba.registry.server.clientInterface.store;


import io.kiki.sba.registry.common.model.Tuple;
import io.kiki.sba.registry.common.model.dataserver.DatumVersion;
import io.kiki.sba.registry.common.model.store.Subscriber;
import io.kiki.sba.registry.server.clientInterface.registry.ClientInterfaceRegistry.SelectSubscriber;
import io.kiki.sba.registry.util.StringUtil;
import org.apache.commons.lang.StringUtils;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.stream.Collectors;


public class SubscribersImpl extends AbstractDataManager<Subscriber> implements Subscribers {

    private final Store<Subscriber> store = new _store_<>(1024 * 16, 256);

    public SubscribersImpl() {
        super();
    }

    @Override
    public boolean add(Subscriber subscriber) {
        StringUtil.nullValue(subscriber.getScope(), "subscriber.scope");
        StringUtil.nullValue(subscriber.getClientVersion(), "subscriber.clientVersion");

        Subscriber.internSubscriber(subscriber);
        Tuple<Subscriber, Boolean> tuple = addData(subscriber);
        return tuple.o2;
    }

    @Override
    public SubscriberVersionCheck checkInterestVersion(String dataCenter, String datumDataInfoId, long version) {
        List<Subscriber> subscriberList = getSubscriberList(datumDataInfoId);
        if (CollectionUtils.isEmpty(subscriberList)) {
            return SubscriberVersionCheck.NoSub;
        }
        for (Subscriber subscriber : subscriberList) {
            if (subscriber.checkVersion(dataCenter, version)) {
                return SubscriberVersionCheck.Interested;
            }
        }
        return SubscriberVersionCheck.Obsolete;
    }

    @Override
    public List<Subscriber> getSubscriberList(String dataInfoId) {
        return getDataList(dataInfoId);
    }

    @Override
    public SelectSubscriber selectSubscribers(Set<String> dataCenterSet) {
        final String localDataCenter = config.getDataCenter();

        Store<Subscriber> store = getSlotStore();
        int dataInfoIdSize = store.getDataInfoIds().size();
        final Map<String, Map<String, DatumVersion>> versions = new HashMap<>(dataCenterSet.size());
        final Map<String, List<Subscriber>> dataCenterToToPushEmptySubscriberListMap = new HashMap<>(dataCenterSet.size());
        final List<Subscriber> toRegisterMultiSubscribers = new ArrayList<>(128);

        for (String dataCenter : dataCenterSet) {
            versions.put(dataCenter, new HashMap<>(dataInfoIdSize));
            dataCenterToToPushEmptySubscriberListMap.put(dataCenter, new ArrayList<>(256));
        }

        store.forEach((String dataInfoId, Map<String, Subscriber> map) -> {
            if (CollectionUtils.isEmpty(map)) {
                return;
            }
            for (Subscriber subscriber : map.values()) {

                if (!subscriber.hasPushed()) {
                    toRegisterMultiSubscribers.add(subscriber);
                    continue;
                }

                for (String dataCenter : dataCenterSet) {
                    Map<String, DatumVersion> dataInfoIdToDatumVersionMap = versions.get(dataCenter);
                    List<Subscriber> pushEmptySubscriberList = dataCenterToToPushEmptySubscriberListMap.get(dataCenter);

                    final boolean isLocalDataCenter = localDataCenter.equals(dataCenter);
                    // not multi sub and not local dataCenter, not interest the other dataCenter's pub
                    if (!subscriber.isAcceptMulti() && !isLocalDataCenter) {
                        continue;
                    }

                    if (subscriber.isMarkedPushEmpty(dataCenter)) {
                        if (subscriber.needPushEmpty(dataCenter)) {
                            pushEmptySubscriberList.add(subscriber);
                        }
                        continue;
                    }
                    final long pushVersion = subscriber.getPushedVersion(dataCenter);
                    DatumVersion maxVersion = dataInfoIdToDatumVersionMap.computeIfAbsent(dataInfoId, k -> new DatumVersion(0));
                    if (maxVersion.getValue() < pushVersion) {
                        dataInfoIdToDatumVersionMap.put(dataInfoId, new DatumVersion(pushVersion));
                    }
                }
            }
        });
        return new SelectSubscriber(versions, dataCenterToToPushEmptySubscriberListMap, toRegisterMultiSubscribers);
    }

    @Override
    protected Store<Subscriber> getSlotStore() {
        return store;
    }

    @Override
    public Map<String, List<String>> filterIPs(String group, int limit) {
        Map<String, List<String>> map = new HashMap<>(1024);
        forEach((String dataInfoId, Map<String, Subscriber> map1) -> {
            if (CollectionUtils.isEmpty(map1)) {
                return;
            }
            String g = map1.values().stream().findAny().get().getGroup();
            if (!StringUtils.equals(g, group)) {
                return;
            }
            map.put(dataInfoId, map1.values().stream().map(subscriber -> subscriber.getClientUrl().getIp()).limit(Math.max(limit, 0)).collect(Collectors.toList()));
        });
        return map;
    }
}
